package com.itheima.stream;

import cn.hutool.core.convert.Convert;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.Reducer;
import org.springframework.context.annotation.Bean;

//@Configuration
//@Slf4j
public class WordCountsDemo2 {

    /* 在TopicA中每输入一个值求和并写入到TopicB */
    @Bean
    public KTable<Object, Object> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
        KTable<Object, Object> kTable = stream.map(new KeyValueMapper<String, String, KeyValue<?, ?>>() {
            @Override
            public KeyValue<?, ?> apply(String key, String value) {
                // Key为空值
                // Value为输入的值
                return new KeyValue<String, String>("sum", value.toString());
            }
        }).groupByKey()
                .reduce(new Reducer<Object>() {
                    @Override
                    public Object apply(Object x, Object y) {
                        // x为上次累加后的值
                        // y为这次要累加的值
                        return Convert.toStr(Convert.toInt(x) + Convert.toInt(y));
                    }
                });

        kTable.toStream().to("itcast-topic-out");
        return kTable;
    }
}